// $ go run _examples/main.go

package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"github.com/elastic/go-elasticsearch"
	"log"
	"os"
	"strings"
	"sync"

	"github.com/elastic/go-elasticsearch/esapi"
)

type EsServer struct {
	es *elasticsearch.Client
}

func (server *EsServer) initEs() {
	var r map[string]interface{}

	cfg := elasticsearch.Config{
		Addresses: []string{
			"http://127.0.0.1:9200",
		},
	}
	var err error
	server.es, err = elasticsearch.NewClient(cfg)

	es := server.es

	//es, err := elasticsearch.NewDefaultClient()
	if err != nil {
		log.Fatalf("Error creating the client: %s", err)
	}

	// 1. Get cluster info
	//
	res, err := es.Info()
	if err != nil {
		log.Fatalf("Error getting response: %s", err)
	}
	// Check response status
	if res.IsError() {
		log.Fatalf("Error: %s", res.String())
	}
	// Deserialize the response into a map.
	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
		log.Fatalf("Error parsing the response body: %s", err)
	}
	// Print client and server version numbers.
	log.Printf("Client: %s", elasticsearch.Version)
	log.Printf("Server: %s", r["version"].(map[string]interface{})["number"])
	log.Println(strings.Repeat("~", 37))
}

func (server *EsServer) indexData(index, id, data string) {
	// Set up the request object.
	req := esapi.IndexRequest{
		Index:      index,
		Body:       strings.NewReader(data),
		Refresh:    "true",
		DocumentID: id,
	}

	// Perform the request with the client.
	res, err := req.Do(context.Background(), server.es)
	if err != nil {
		log.Fatalf("Error getting response: %s", err)
	}
	defer res.Body.Close()

	if res.IsError() {
		j, _ := json.Marshal(res)
		fmt.Printf(string(j))
		log.Printf("[%s] Error indexing document : %s", res.Status(), data)
	} else {
		// Deserialize the response into a map.
		var r map[string]interface{}
		if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
			log.Printf("Error parsing the response body: %s", err)
		} else {
			// Print the response status and indexed document version.
			log.Printf("[%s] %s; version=%d", res.Status(), r["result"], int(r["_version"].(float64)))
		}
	}
}

func (server EsServer) esSearch(index string, query interface{}) {

	var r map[string]interface{}

	var buf bytes.Buffer

	if err := json.NewEncoder(&buf).Encode(query); err != nil {
		log.Fatalf("Error encoding query: %s", err)
	}

	es := server.es
	// Perform the search request.
	res, err := es.Search(
		es.Search.WithContext(context.Background()),
		es.Search.WithIndex("test"),
		es.Search.WithBody(&buf),
		es.Search.WithTrackTotalHits(true),
		es.Search.WithPretty(),
	)
	if err != nil {
		log.Fatalf("Error getting response: %s", err)
	}
	defer res.Body.Close()

	if res.IsError() {
		var e map[string]interface{}
		if err := json.NewDecoder(res.Body).Decode(&e); err != nil {
			log.Fatalf("Error parsing the response body: %s", err)
		} else {
			// Print the response status and error information.
			log.Fatalf("[%s] %s: %s",
				res.Status(),
				e["error"].(map[string]interface{})["type"],
				e["error"].(map[string]interface{})["reason"],
			)
		}
	}

	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
		log.Fatalf("Error parsing the response body: %s", err)
	}
	// Print the response status, number of results, and request duration.
	log.Printf(
		"[%s] %d hits; took: %dms",
		res.Status(),
		int(r["hits"].(map[string]interface{})["total"].(map[string]interface{})["value"].(float64)),
		int(r["took"].(float64)),
	)
	// Print the ID and document source for each hit.
	for _, hit := range r["hits"].(map[string]interface{})["hits"].([]interface{}) {
		log.Printf(" * ID=%s, %s", hit.(map[string]interface{})["_id"], hit.(map[string]interface{})["_source"])
	}

	log.Println(strings.Repeat("=", 37))
}

func main() {
	log.SetFlags(0)

	var wg sync.WaitGroup

	var server EsServer
	server.initEs()

	// 2. Index documents concurrently
	//
	file, err := os.Open("E:\\A_SW\\topic-daily-equal-data.txt")
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {

		lineText := scanner.Text()
		dataItem := strings.Split(lineText, "(|)")

		var index string
		switch dataItem[0] {
		case "SCENIC":
			index = "lbs_21q1v1_detail_scenic"
		case "GAS":
			index = "lbs_21q1v1_detail_gasstation"
		case "CATER":
			index = "lbs_21q1v1_detail_cater"
		default:
			fmt.Printf(dataItem[0])
		}

		//go func() {
		server.indexData(index, dataItem[2], dataItem[4])
		//wg.Done()
		//}()
	}
	wg.Wait()

	log.Println(strings.Repeat("-", 37))

	// 3. Search for the indexed documents
	//
	// Build the request body.

	//query := map[string]interface{}{
	//	"query": map[string]interface{}{
	//		"match": map[string]interface{}{
	//			"title": "test",
	//		},
	//	},
	//}

}
